package com.ksyun.campus.client;


import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ksyun.campus.client.common.constant.ServerConfiguration;
import com.ksyun.campus.client.domain.BaseResult;
import com.ksyun.campus.client.domain.ReplicaData;
import com.ksyun.campus.client.domain.StatInfo;
import com.ksyun.campus.client.domain.dto.WriteReqDto;
import com.ksyun.campus.client.util.HttpClientConfig;
import com.ksyun.campus.client.util.HttpClientUtil;
import lombok.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.hc.client5.http.classic.HttpClient;
import org.apache.hc.client5.http.classic.methods.HttpPost;
import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
import org.apache.hc.core5.http.ContentType;
import org.apache.hc.core5.http.HttpStatus;
import org.apache.hc.core5.http.io.entity.EntityUtils;
import org.apache.hc.core5.http.io.entity.StringEntity;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;

@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
@Slf4j
public class FSOutputStream extends OutputStream {
    private static final int BUFFER_LIMIT = 2 * 1024 * 1024; // 2MB
//    private static final int BUFFER_LIMIT = 20 * 1024 * 1024; // 20MB
    private StatInfo statInfo;
    private String path;
    private String defaultFileSystemName;
    private String metaServerUrl;
    private ByteArrayOutputStream buffer = new ByteArrayOutputStream();
    private boolean closed = false;

    @Override
    public void write(int b) throws IOException {
        buffer.write(b);
        if (buffer.size() >= BUFFER_LIMIT) {
            flushToDataServer();
        }
    }

    @Override
    public void write(byte[] b) throws IOException {
        write(b, 0, b.length);
    }

    @Override
    public void write(byte[] b, int off, int len) throws IOException {
        int remainingData = len;

        try {
            while (remainingData > 0) {
                int spaceInBuffer = BUFFER_LIMIT - buffer.size();
                int bytesToWrite = Math.min(spaceInBuffer, remainingData);

                buffer.write(b, off, bytesToWrite);

                remainingData -= bytesToWrite;
                off += bytesToWrite;

                if (buffer.size() >= BUFFER_LIMIT) {
                    flushToDataServer();
                }
            }
        } catch (Exception e) {
            commitToMetaServer();
            throw e;
        }
    }

    @Override
    public void close() throws IOException {
        if (!closed) {
            closed = true;

            // 将剩余的数据写入 DataServer
            if (buffer.size() > 0) {
                flushToDataServer();
            }

            // 提交写请求给 MetaServer
            commitToMetaServer();

            buffer.close();
            super.close();
        }
    }

    /**
     * 将缓冲区的文件数据发送给DataServer
     * @throws IOException
     */
    private void flushToDataServer() throws IOException {
        byte[] data = buffer.toByteArray();
        ReplicaData replicaData = statInfo.getReplicaData().get(0);
        String ipAndPort = replicaData.getDsNode();
        String dataServerUrl = ServerConfiguration.HTTP + ipAndPort + ServerConfiguration.DATAWRITEPATH;

        HttpPost dataPost = new HttpPost(dataServerUrl);
        dataPost.addHeader("fileSystemName", defaultFileSystemName);
        ContentType contentType = ContentType.APPLICATION_JSON;

        WriteReqDto writeReqDto = new WriteReqDto(statInfo, data, 0, data.length);

        // 创建 StringEntity 并设置到 HttpPost 请求中
        ObjectMapper objectMapper = new ObjectMapper();
        String jsonReqBody = objectMapper.writeValueAsString(writeReqDto);
        StringEntity entity = new StringEntity(jsonReqBody, contentType);
        dataPost.setEntity(entity);

        HttpClient clientToDS = HttpClientUtil.createHttpClient(new HttpClientConfig());

        try (CloseableHttpResponse responseDS = (CloseableHttpResponse) clientToDS.execute(dataPost)) {
            if (responseDS.getCode() == HttpStatus.SC_OK) {
                String resBodyDS = EntityUtils.toString(responseDS.getEntity());
                BaseResult<StatInfo> result = new ObjectMapper().readValue(resBodyDS,
                        new TypeReference<BaseResult<StatInfo>>() {
                        });

                if (result.getCode() == 200) {
                    StatInfo dataStatInfo = result.getData();
                    dataStatInfo.setSize(statInfo.getSize() + data.length);
//                    System.out.println("更新元数据大小" + statInfo.getSize());
                    statInfo = dataStatInfo;
                } else {
//                    System.out.println("DataServer写数据失败: " + result.getMsg());/
                    //向MetaServer发送提交写成功与否 失败:statInfo.size=-1
                    statInfo.setSize(-1);
                    commitToMetaServer();
                    throw new IOException("Failed to write to DataServer, unexpected result code: " + result.getCode());
                }
            } else {
                throw new IOException("Failed to write to DataServer with HTTP status code: " + responseDS.getCode());
            }
        } catch (Exception e) {
//            log.error("Failed to write to DataServer", e);
            throw new IOException("Failed to write to DataServer", e);
        } finally {
            resetBuffer();
        }
    }

    private void resetBuffer() {
        buffer.reset();
    }

    private void commitToMetaServer() throws IOException {
        sendMetadataToMetaServer(statInfo);
    }

    /**
     * 向MetaServer发送写入文件完成后的元数据信息
     * @param statInfo
     * @throws IOException
     */
    private void sendMetadataToMetaServer(StatInfo statInfo) throws IOException {
        String url = metaServerUrl + ServerConfiguration.METAWRITEPATH;
        HttpPost httpPost = new HttpPost(url);
        httpPost.addHeader("fileSystemName", defaultFileSystemName);

        WriteReqDto writeReqDtoMS = new WriteReqDto(statInfo, null, 0, 0);
        ObjectMapper objectMapper = new ObjectMapper();
        String jsonReqBodyMS = objectMapper.writeValueAsString(writeReqDtoMS);
        StringEntity entityMS = new StringEntity(jsonReqBodyMS, ContentType.APPLICATION_JSON);
        httpPost.setEntity(entityMS);

        HttpClient clientToMS = HttpClientUtil.createHttpClient(new HttpClientConfig());

        try (CloseableHttpResponse responseMS = (CloseableHttpResponse) clientToMS.execute(httpPost)) {
            if (responseMS.getCode() == HttpStatus.SC_OK) {
                String resBodyMS = EntityUtils.toString(responseMS.getEntity());
                BaseResult<String> result = new ObjectMapper().readValue(resBodyMS,
                        new TypeReference<BaseResult<String>>() {
                        });
                if (result.getCode() == 200) {
//                    System.out.println("成功向MetaServer提交写入, metaServer返回的信息:" + result.getData());
                } else {
//                    System.out.println("失败向MetaServer提交写入, metaServer返回的信息:" + result.getMsg());
                    throw new IOException("Failed to write to DataServer, unexpected result code: " + result.getCode());
                }
            } else {
                throw new IOException("Failed to commit write to MetaServer with HTTP status code: " + responseMS.getCode());
            }
        } catch (Exception eMS) {
            eMS.printStackTrace();
            throw new IOException("Failed to commit write to MetaServer", eMS);
        }
    }
}


//import com.fasterxml.jackson.core.type.TypeReference;
//import com.fasterxml.jackson.databind.ObjectMapper;
//import com.ksyun.campus.client.common.constant.ServerConfiguration;
//import com.ksyun.campus.client.common.context.ServerContext;
//import com.ksyun.campus.client.domain.BaseResult;
//import com.ksyun.campus.client.domain.ReplicaData;
//import com.ksyun.campus.client.domain.StatInfo;
//import com.ksyun.campus.client.domain.dto.WriteReqDto;
//import com.ksyun.campus.client.util.HttpClientConfig;
//import com.ksyun.campus.client.util.HttpClientUtil;
//import lombok.*;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.hc.client5.http.classic.HttpClient;
//import org.apache.hc.client5.http.classic.methods.HttpPost;
//import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
//import org.apache.hc.core5.http.ContentType;
//import org.apache.hc.core5.http.HttpStatus;
//import org.apache.hc.core5.http.io.entity.EntityUtils;
//import org.apache.hc.core5.http.io.entity.StringEntity;
//
//import java.io.ByteArrayOutputStream;
//import java.io.IOException;
//import java.io.OutputStream;
//import java.util.concurrent.CompletableFuture;
//
//@Data
//@AllArgsConstructor
//@NoArgsConstructor
//@Builder
//@Slf4j
//public class FSOutputStream extends OutputStream {
//    private StatInfo statInfo;
//    private String path;
//    private String defaultFileSystemName;
//    private  String metaServerUrl;
//
//    // 内存缓存用于暂存数据
//    private ByteArrayOutputStream buffer;
//
//    private boolean closed = false;
//
//    @Override
//    public void write(int b) throws IOException {
//        buffer.write(b);
//    }
//
//    @Override
//    public void write(byte[] b) throws IOException {
//        buffer.write(b, 0, b.length);
//    }
//
//    @Override
//    public void write(byte[] b, int off, int len) throws IOException {
//        buffer.write(b, off, len);
//    }
//
//    @Override
//    public void close() throws IOException {
//        if (!closed) {
//            closed = true;
//
//            // 将数据异步写入 DataServer
//            CompletableFuture.runAsync(() -> {
//                try {
//                    flushToDataServer();
//                } catch (IOException e) {
//                    System.out.println("Failed to flush data to DataServer: \n"+e);
//                }
//            }).join();  // 如果需要阻塞直到任务完成，使用 join，否则移除它以异步执行
//
//            buffer.close();
//            super.close();
//        }
//    }
//
//    /**
//     * 将缓冲区的文件数据发送给DataServer (在client修改StatInfo中size大小)
//     * @throws IOException
//     */
//    private void flushToDataServer() throws IOException {
//        // 获取要写入的字节数据
//        byte[] data = buffer.toByteArray();
//        int len = data.length;
//        // 与 DataServer 通信写入实际数据
//        ReplicaData replicaData = statInfo.getReplicaData().get(0);
//        String ipAndPort = replicaData.getDsNode();
//        String dataServerUrl = ServerConfiguration.HTTP + ipAndPort + ServerConfiguration.DATAWRITEPATH;
//
//        HttpPost dataPost = new HttpPost(dataServerUrl);
//        dataPost.addHeader("fileSystemName", defaultFileSystemName);
//        ContentType contentType = ContentType.APPLICATION_JSON;
//
//        WriteReqDto writeReqDto = new WriteReqDto(statInfo, data, 0, len);
//
//        // 创建 StringEntity 并设置到 HttpPost 请求中
//        ObjectMapper objectMapper = new ObjectMapper();
//        String jsonReqBody = objectMapper.writeValueAsString(writeReqDto);
//        StringEntity entity = new StringEntity(jsonReqBody, contentType);
//        dataPost.setEntity(entity);
//
//        HttpClient ClientToDS = HttpClientUtil.createHttpClient(new HttpClientConfig());
//        try (CloseableHttpResponse responseDS = (CloseableHttpResponse) ClientToDS.execute(dataPost)) {
//            if (responseDS.getCode() == HttpStatus.SC_OK) {
//                String resBodyDS = EntityUtils.toString(responseDS.getEntity());
//                BaseResult<StatInfo> result = new ObjectMapper().readValue(resBodyDS,
//                        new TypeReference<BaseResult<StatInfo>>() {});
//
//                if (result.getCode() == 200) {
//
//                    StatInfo dataStatInfo = result.getData();
//                    dataStatInfo.setSize(len);
//                    System.out.println("发给metaServer的元数据: "+dataStatInfo);
//                    // 将元数据信息发送给 MetaServer
//                    sendMetadataToMetaServer(dataStatInfo,0, len);
//                } else {
//                    System.out.println("DataServer写数据失败: "+result.getMsg());
//                    //向MetaServer发送提交写成功与否 失败:statInfo.size=-1
//                    statInfo.setSize(-1);
//                    sendMetadataToMetaServer(statInfo,0,0);
//                    throw new IOException("Failed to write to DataServer, unexpected result code: " + result.getCode());
//                }
//            } else {
//                throw new IOException("Failed to write to DataServer with HTTP status code: " + responseDS.getCode());
//            }
//        } catch (Exception e) {
//            throw new IOException("Failed to write to DataServer", e);
//        }
//    }
//
//    /**
//     * 向MetaServer发送写入文件完成后的元数据信息
//     * @param statInfo
//     * @param off
//     * @param len
//     * @throws IOException
//     */
//    private void sendMetadataToMetaServer(StatInfo statInfo,  int off, int len) throws IOException {
//        String url = metaServerUrl + ServerConfiguration.METAWRITEPATH;
//        HttpPost httpPost = new HttpPost(url);
//        httpPost.addHeader("fileSystemName", defaultFileSystemName);
//
//        WriteReqDto writeReqDtoMS = new WriteReqDto(statInfo, null, off, len);
//        ObjectMapper objectMapper = new ObjectMapper();
//        String jsonReqBodyMS = objectMapper.writeValueAsString(writeReqDtoMS);
//        StringEntity entityMS = new StringEntity(jsonReqBodyMS, ContentType.APPLICATION_JSON);
//        httpPost.setEntity(entityMS);
//
//        HttpClient ClientToMS = HttpClientUtil.createHttpClient(new HttpClientConfig());
//
//        // 执行请求并处理响应
//        try (CloseableHttpResponse responseMS = (CloseableHttpResponse) ClientToMS.execute(httpPost)) {
//            if (responseMS.getCode() == HttpStatus.SC_OK) {
//                String resBodyMS = EntityUtils.toString(responseMS.getEntity());
//                BaseResult<String> result = new ObjectMapper().readValue(resBodyMS,
//                        new TypeReference<BaseResult<String>>() {});
//                if(result.getCode()==200){
//                    System.out.println("成功向MetaServer提交写入, metaServer返回的信息:"+result.getData());
//                }else {
//                    System.out.println("失败向MetaServer提交写入, metaServer返回的信息:"+result.getMsg());
//                    throw new IOException("Failed to write to DataServer, unexpected result code: "+ result.getCode());
//                }
//            } else {
//                throw new IOException("Failed to commit write to MetaServer with HTTP status code: " + responseMS.getCode());
//            }
//        } catch (Exception eMS) {
//            eMS.printStackTrace();
//            throw new IOException("Failed to commit write to MetaServer", eMS);
//        }
//    }
//
//}